//package kn.pulldata.mq;
//
//
//import cn.hutool.crypto.digest.DigestUtil;
//import cn.hutool.http.HttpStatus;
//import cn.hutool.json.JSONUtil;
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONException;
//import com.alibaba.fastjson.JSONObject;
//import com.alibaba.fastjson.serializer.SerializerFeature;
//import com.rabbitmq.client.Channel;
//import kn.pulldata.common.OkHttpUtils;
//import kn.pulldata.common.RabbitMqCon;
//import kn.pulldata.common.ThreadPoolUtil;
//import kn.pulldata.req.QueryTonalStateReq;
//import kn.pulldata.resp.MatchDistanceResp;
//import kn.pulldata.resp.PlanManagesItem;
//import kn.pulldata.service.ExternalService;
//import kn.pulldata.service.RedisLockService;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//import java.util.ArrayList;
//import java.util.List;
//import java.util.Objects;
//import java.util.concurrent.CompletableFuture;
//import java.util.stream.Collectors;
//
//@Component
//@Slf4j
//public class PushConsumer2 {
//
//    @Value("${monitor.keyword.saveUrl}")
//    private String getKeywordSaveUrl;
//
//    @Autowired
//    private ExternalService externalService;
//
//    @Autowired
//    private RedisLockService redisLockService;
//
//
//    /*@RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE,concurrency = "20")
//    public void keywordReceiveMessage(String message, Channel channel, Message rabbitMessage) {
//
//        try {
//            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
//            if(matchDistanceRespList.isEmpty()){
//                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
//                return;
//            }
//            ThreadPoolUtil.execute(() ->{
//                push(matchDistanceRespList);
//            });
//            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
//
//        }
//        catch (JSONException jsonException){
//            log.error("json解析异常，消息: {}" ,JSONUtil.toJsonStr(message));
//        }catch (Exception e) {
//            e.printStackTrace();
//            try {
//                int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
//                if (retryCount > 5) {
//                    // 重试次数超过阈值，选择丢弃
//                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
//                    log.error("消息推送数据重试次数超过阈值，丢弃消息: {}" + rabbitMessage.getMessageProperties().getMessageId());
//                } else {
//                    // 否则继续重试
//                    channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
//                }
//            } catch (IOException ex) {
//                ex.printStackTrace();
//            }
//        }
//    }
//
//    private void push(List<MatchDistanceResp> matchDistanceRespList) {
//
//        // 处理单个请求的逻辑
//        long l = System.currentTimeMillis();
//        List<CompletableFuture<MatchDistanceResp>> futures = matchDistanceRespList.stream()
//                .map(resp -> CompletableFuture.supplyAsync(() -> singleRequest(resp), ThreadPoolUtil.getExecutor()))
//                .collect(Collectors.toList());
//        CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
//        allOf.thenRun(() -> {
//            List<MatchDistanceResp> resps = futures.stream()
//                    .map(CompletableFuture::join)
//                    .filter(Objects::nonNull)
//                    .collect(Collectors.toList());
//
//            if (!resps.isEmpty()) {
//                String req = JSON.toJSONString(resps, SerializerFeature.WriteNullNumberAsZero, SerializerFeature.WriteNullListAsEmpty);
//                long start = System.currentTimeMillis();
//                if (pullData(req, getKeywordSaveUrl)) {
//                    log.info("推送数据成功：{}",JSONUtil.toJsonStr(resps.stream().map(MatchDistanceResp::getDataID).collect(Collectors.toList())));
//                }else {
//                    log.error("推送数据重试次数超过后异常：{}",JSONUtil.toJsonStr(req));
//                }
//                long end = System.currentTimeMillis();
//                log.warn("推送数据耗时：{}毫秒",(end - start));
//            }
//        });
//
//    }
//
//
//    private MatchDistanceResp singleRequest(MatchDistanceResp resp) {
//        //去重
//        try {
//            List<PlanManagesItem> planManagesItemList = new ArrayList<>();
//
//            List<String> keyList = new ArrayList<>();
//            for (PlanManagesItem planManage : resp.getPlanManages()) {
//                String key = planManage.getPlanManageId() + resp.getWebsite();
//                String uniqueKeyHash = DigestUtil.md5Hex(key);
//                keyList.add(uniqueKeyHash);
//            }
//            long start = System.currentTimeMillis();
//
//            List<Boolean> lockResults = redisLockService.obtainLocksInBatch(keyList, 3600 * 72);
//            long end = System.currentTimeMillis();
//            log.warn("上锁耗时：{}毫秒",(end - start));
//            for (int i = 0; i < lockResults.size(); i++) {
//                if (lockResults.get(i)) {
//                    planManagesItemList.add(resp.getPlanManages().get(i));
//                }
//            }
//            if(planManagesItemList.isEmpty()){
//                return null;
//            }
//            resp.setPlanManages(planManagesItemList);
//        }catch (Exception e){
//            e.printStackTrace();
//            log.error("去重出错：{},异常信息{}",resp, e.getMessage());
//        }
//        QueryTonalStateReq queryTonalStateReq = new QueryTonalStateReq();
//        queryTonalStateReq.setTitle(resp.getTitle());
//        queryTonalStateReq.setContent(resp.getContent() + " " + resp.getImageContent());
//        long start = System.currentTimeMillis();
//        //String tonalState = externalService.queryTonalState(queryTonalStateReq);
//        long end = System.currentTimeMillis();
//        log.warn("调性耗时：{}毫秒",(end - start));
//        resp.setTonalState("1");
//        return resp;
//    }
//
//    private boolean pullData(String req, String url){
//        // 处理消息逻辑
//        int maxRetries = 3; // 最大重试次数
//        int attempts = 0; // 当前尝试次数
//        boolean success = false;
//
//        while (attempts < maxRetries) {
//            try {
//                String body = OkHttpUtils.postJson(url, req);
//                cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(body);
//                if (HttpStatus.HTTP_OK == jsonObject.getInt("code")) {
//                    success = true; // 请求成功，退出重试
//                    log.info("推送成功！！！！！！！！！");
//                    return success;
//                } else {
//                    log.error("推送失败");
//                    attempts++;
//
//                    // 增加等待时间，避免频繁请求
//                    try {
//                        Thread.sleep(1000); // 等待1秒后重试
//                    } catch (InterruptedException e) {
//                        Thread.currentThread().interrupt(); // 恢复中断状态
//                    }
//                }
//            } catch (Exception e) {
//                e.printStackTrace();
//                throw new RuntimeException("推送数据异常:"+ e.getMessage());
//            }
//
//        }
//        return false;
//    }*/
//
//    @RabbitListener(queues = RabbitMqCon.PUSH_MATCHED_DATA_QUEUE,concurrency = "20")
//    public void keywordReceiveMessage(String message, Channel channel, Message rabbitMessage) {
//        try {
//            List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
//            if (matchDistanceRespList.isEmpty()) {
//                channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
//                return;
//            }
//
//            // 异步处理
//            CompletableFuture.runAsync(() -> push(matchDistanceRespList), ThreadPoolUtil.getExecutor())
//                    .whenComplete((result, throwable) -> {
//                        try {
//                            channel.basicAck(rabbitMessage.getMessageProperties().getDeliveryTag(), false);
//                        } catch (IOException e) {
//                            log.error("消息确认失败", e);
//                        }
//                    });
//
//        } catch (JSONException jsonException) {
//            log.error("json解析异常，消息: {}", JSONUtil.toJsonStr(message));
//        } catch (Exception e) {
//            handleConsumerException(e, rabbitMessage, channel);
//        }
//    }
//    private MatchDistanceResp singleRequest(MatchDistanceResp resp) {
//        //去重
//        try {
//            List<PlanManagesItem> planManagesItemList = new ArrayList<>();
//
//            List<String> keyList = new ArrayList<>();
//            for (PlanManagesItem planManage : resp.getPlanManages()) {
//                String key = planManage.getPlanManageId() + resp.getWebsite();
//                String uniqueKeyHash = DigestUtil.md5Hex(key);
//                keyList.add(uniqueKeyHash);
//            }
//            long start = System.currentTimeMillis();
//
//            List<Boolean> lockResults = redisLockService.obtainLocksInBatch(keyList, 3600 * 72);
//            long end = System.currentTimeMillis();
//            log.warn("上锁耗时：{}毫秒",(end - start));
//            for (int i = 0; i < lockResults.size(); i++) {
//                if (lockResults.get(i)) {
//                    planManagesItemList.add(resp.getPlanManages().get(i));
//                }
//            }
//            if(planManagesItemList.isEmpty()){
//                return null;
//            }
//            resp.setPlanManages(planManagesItemList);
//        }catch (Exception e){
//            e.printStackTrace();
//            log.error("去重出错：{},异常信息{}",resp, e.getMessage());
//        }
//        QueryTonalStateReq queryTonalStateReq = new QueryTonalStateReq();
//        queryTonalStateReq.setTitle(resp.getTitle());
//        queryTonalStateReq.setContent(resp.getContent() + " " + resp.getImageContent());
//        long start = System.currentTimeMillis();
//        //String tonalState = externalService.queryTonalState(queryTonalStateReq);
//        long end = System.currentTimeMillis();
//        log.warn("调性耗时：{}毫秒",(end - start));
//        resp.setTonalState("1");
//        return resp;
//    }
//    private void push(List<MatchDistanceResp> matchDistanceRespList) {
//        List<CompletableFuture<MatchDistanceResp>> futures = matchDistanceRespList.stream()
//                .map(resp -> CompletableFuture.supplyAsync(() -> singleRequest(resp), ThreadPoolUtil.getExecutor()))
//                .collect(Collectors.toList());
//
//        // 等待所有请求完成
//        CompletableFuture<Void> allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
//        allOf.thenRun(() -> {
//            List<MatchDistanceResp> resps = futures.stream()
//                    .map(CompletableFuture::join)
//                    .filter(Objects::nonNull)
//                    .collect(Collectors.toList());
//
//            if (!resps.isEmpty()) {
//                String req = JSON.toJSONString(resps, SerializerFeature.WriteNullNumberAsZero, SerializerFeature.WriteNullListAsEmpty);
//                // 异步执行数据推送
//                CompletableFuture.runAsync(() -> pullData(req, getKeywordSaveUrl), ThreadPoolUtil.getExecutor());
//            }
//        }).exceptionally(e -> {
//            log.error("处理过程中出现异常: {}", e.getMessage());
//            return null;
//        });
//    }
//
//    private boolean pullData(String req, String url) {
//        // 使用异步 HTTP 客户端
//        return CompletableFuture.supplyAsync(() -> {
//            int maxRetries = 3;
//            int attempts = 0;
//            boolean success = false;
//
//            while (attempts < maxRetries) {
//                try {
//                    long start = System.currentTimeMillis();
//                    String body = OkHttpUtils.postJson(url, req);
//                    cn.hutool.json.JSONObject jsonObject = JSONUtil.parseObj(body);
//                    if (HttpStatus.HTTP_OK == jsonObject.getInt("code")) {
//                        long end = System.currentTimeMillis();
//                        log.info("推送成功！耗时:{}",end -start);
//                        return true;
//                    } else {
//                        log.error("推送失败:{}",JSONUtil.toJsonStr(body));
//                        attempts++;
//                        Thread.sleep(1000); // 等待后重试
//                    }
//                } catch (Exception e) {
//                    log.error("推送数据异常: {}", e.getMessage());
//                    break;
//                }
//            }
//            return false;
//        }, ThreadPoolUtil.getExecutor()).join(); // 等待结果
//    }
//
//    private void handleConsumerException(Exception e, Message rabbitMessage, Channel channel) {
//        try {
//            int retryCount = (int) rabbitMessage.getMessageProperties().getHeaders().getOrDefault("x-death", 0);
//            if (retryCount > 5) {
//                channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, false);
//                log.error("消息重试次数超过阈值，丢弃消息: {}", rabbitMessage.getMessageProperties().getMessageId());
//            } else {
//                channel.basicNack(rabbitMessage.getMessageProperties().getDeliveryTag(), false, true);
//            }
//        } catch (IOException ex) {
//            log.error("消息处理异常", ex);
//        }
//    }
//
//
//
//    //**********************************测试推送队列**********************************//
//    //@RabbitListener(queues = RabbitMqCon.KEYWORD_TEST_PUSH_QUEUE)
//    public void testMessage(String message, Channel channel, Message rabbitMessage) {
//
//        List<MatchDistanceResp> matchDistanceRespList = JSONObject.parseArray(message, MatchDistanceResp.class);
//        log.info("测试推送队列监听到消息:{},个数：{}" , rabbitMessage.getMessageProperties().getMessageId(),matchDistanceRespList.size());
//
//        //keywordReceiveMessage(message, channel, rabbitMessage);
//
//    }
//
//
//}
